package com.fdb.efp.nls.common.util;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.*;

/**
 *
 * 功能描述：<br/>
 * 创建时间：2020/4/13 19:09
 * @Author dengqiwen
 * @Version 1.0
 * @Since 1.0
 */
public class KafkaUtil {
    private static Logger logger = LoggerFactory.getLogger(KafkaUtil.class);


    /*kafka主机ip:端口号*/
    private String servers;
    /*kafka发送的主题topic*/
    private String kafkaToptic;


    private static ProducerRecord<String,String> record;

    /**
     * 默认构造方法
     */
    public KafkaUtil(String servers, String kafkaToptic) {
        this.servers=servers;
        this.kafkaToptic=kafkaToptic;
    }


    public  void sendMessage( String kafkaJson) throws Exception{
        logger.info("KAFKA开始发送信息: " + "servers=" + servers + ", " + "kafkaToptic=" + kafkaToptic + ", " + "kafkaJson=" + kafkaJson);
        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");//键序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//值序列化
        props.put("max.block.ms", "6000");//当生产者的发送缓冲区已满，或者没有可用的元数据时，这些方法会阻塞。在阻塞时间达到max.block.ms 时，生产者会抛出超时异常。阻塞超时时间6s
        KafkaProducer<String,String> producer = new KafkaProducer<String,String>(props);
        try {
            record = new ProducerRecord<>(kafkaToptic,kafkaJson);
            producer.send(record).get(60000, TimeUnit.MILLISECONDS);//同步发送，阻塞超时时间6s
            logger.info("KAFKA发送信息成功: " + "servers=" + servers + ", " + "kafkaToptic=" + kafkaToptic + ", " + "kafkaJson=" + kafkaJson);
        } catch (Exception e) {
            logger.error("KAFKA发送信息失败，异常信息：" + e.getMessage());
            e.printStackTrace();
            throw e;
        } finally {
            //关闭生产者，因为生产者会等消息发送成功才关闭，所以设置阻塞超时时间6s
            producer.close(Duration.ofMillis(60000));
        }
    }


}
